/*******************************************************************************
 * Copyright 2010 DemandSpot
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 ******************************************************************************/
package com.demandspot.aws.sqs;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.log4j.Logger;

import com.demandspot.aws.sqs.SqsMessageHandler.Action;
import com.xerox.amazonws.sqs2.Message;
import com.xerox.amazonws.sqs2.MessageQueue;
import com.xerox.amazonws.sqs2.SQSException;


class SqsMessagePoller implements Runnable {
	private static final Logger logger = Logger.getLogger(SqsMessagePoller.class);
	
	private static enum HandlerMode {SINGLE, BATCH};

	private static int SLEEP_TIME = 5000;
	private SqsMessageHandler messageHandler;
	private String queueName;
	private MessageQueue messageQueue;
	private int numOfMessages;
	private boolean running = false;
	private HandlerMode handlerMode;

	private int hideInSecs;

	SqsMessagePoller(MessageQueue messageQueue, SqsMessageHandler messageHandler, String queueName, int numOfMessages, int hideInSecs) {
		super();
		this.messageQueue = messageQueue;
		this.queueName = queueName;
		this.messageHandler = messageHandler;
		this.numOfMessages = numOfMessages;
		this.hideInSecs = hideInSecs;
		
		if (messageHandler instanceof SqsBatchMessageHandler) {
			handlerMode = HandlerMode.BATCH;
			((SqsBatchMessageHandler) messageHandler).setMessageQueue(messageQueue);
		} else {
			handlerMode = HandlerMode.SINGLE;
		}
 		
	}
	
	private void pollBatch() {
		logger.debug("Polling SQS for messages " + handlerMode);
		if (!messageHandler.doBeforePoll()) {
			logger.debug("In messageHandler.doBeforePoll()");
			return;
		}

		SqsBatchMessageHandler sqsBatchMessageHandler = (SqsBatchMessageHandler) messageHandler;
		List<Message> messagesList = new ArrayList<Message>();
		Message[] messages = null;
		try {
			while(true) {
				try {
					messages = messageQueue.receiveMessages(numOfMessages,hideInSecs > 0 ? hideInSecs : 60);
					if (logger.isDebugEnabled()) {
						logger.debug("Got " + (messages == null ? 0 : messages.length) + " from " + messageQueue.getUrl());
					}
				} catch (SQSException e) {
					logger.warn("Failed to receive messages from: " + queueName, e);
					sleep();
					continue;
				}
	
				if (messages != null && messages.length > 0) {
					messagesList.addAll(Arrays.asList(messages));
				}
				
				if (messagesList.size() >= sqsBatchMessageHandler.batchSize) {
					sqsBatchMessageHandler.handleMessages(messagesList);
					messagesList.clear();
				}
				
				if (messages == null || messages.length == 0) {
					sqsBatchMessageHandler.handleMessages(messagesList);
					messagesList.clear();
					setRunning(false);
					break;
				}
			}
		} catch (Throwable t) {
			logger.warn("handler " + messageHandler.getClass() + " failed", t);
		} finally {
			setRunning(false);
		}
	}
	
	private void pollSingle() {
		logger.debug("Polling SQS for messages " + handlerMode);
		if (!messageHandler.doBeforePoll()) {
			logger.debug("In messageHandler.doBeforePoll()");
			return;
		}

		Message[] messages = null;
		try {
			while(true) {
				try {
					if (hideInSecs > 0) {
						messages = messageQueue.receiveMessages(numOfMessages, hideInSecs);
					} else {
						messages = messageQueue.receiveMessages(numOfMessages);
					}
					if (logger.isDebugEnabled()) {
						logger.debug("Got " + (messages == null ? 0 : messages.length) + " from " + messageQueue.getUrl());
					}
				} catch (SQSException e) {
					logger.warn("Failed to receive messages from: " + queueName, e);
					sleep();
					continue;
				}
	
				if (messages != null) {
					for (Message message : messages) {
						try {
							Action action = messageHandler.handleMessage(message);
	
							if (action == Action.DeleteMessage) {
								messageQueue.deleteMessage(message);
							}
						} catch (Exception e) {
							logger.error("Error handling message: " + message.getMessageId() + ", queue: " + queueName, e);
						}
					}
				}
	
				if (messages == null || messages.length == 0) {
					messageHandler.onQueueEmpty();
					setRunning(false);
					break;
				}
			}
		} catch (Throwable t) {
			logger.warn("Handler " + messageHandler.getClass() + " failed", t);
		} finally {
			setRunning(false);
		}
	}
	
	synchronized boolean isRunning() {
		return running;
	}
	
	synchronized void setRunning(boolean running) {
		this.running = running;
	}

	private void sleep() {
		try {
			Thread.sleep(SLEEP_TIME);
		} catch (InterruptedException e) {
		}
	}

	@Override
	public void run() {
		switch (handlerMode) {
		case BATCH:
			pollBatch();
			break;
		case SINGLE:
			pollSingle();
			break;
		}
	}

	@Override
	public String toString() {
		return "SqsMessagePoller [queueName=" + queueName + "]";
	}
	
}
